package io.github.junxworks.qt.commons.extractor;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Component;

import com.google.common.collect.Maps;

import io.github.junxworks.junx.core.exception.BaseRuntimeException;
import io.github.junxworks.junx.core.util.StringUtils;
import io.github.junxworks.junx.event.EventBus;
import io.github.junxworks.junx.event.EventContext;
import io.github.junxworks.qt.config.Params;

@Component("DBDataExtractor")
public class DBDataExtractor implements DataExtractor {

	private Logger log = LoggerFactory.getLogger(DBDataExtractor.class);

	@Autowired
	private EventBus eventBus;

	@Autowired
	private NamedParameterJdbcTemplate jdbcTemplate;

	@Override
	public void extract(ExtractContext context) {
		String topic = context.getTopic();
		if (StringUtils.isNull(topic)) {
			throw new BaseRuntimeException("Topic can not be empty");
		}
		String sql = null;
		Map<String, Object> params = Maps.newHashMap();
		int capacity = context.getCapacity();
		if (capacity == -1) {
			sql = "select count(1) from (" + context.getSql() + ") a";
			Integer count = jdbcTemplate.queryForObject(sql, params, Integer.class);
			if (count == 0) {
				return;
			}
			capacity = count;
		}
		log.info(topic + "数据抽取事件共" + capacity + "条记录待发送");
		int pages = capacity / context.getPageSize();
		int pageSize = context.getPageSize();
		String paramName = context.getObjectName();
		for (int i = 0, len = pages + 1; i < len; i++) {
			sql = context.getSql() + " limit " + (i * pageSize) + "," + pageSize;
			List<Map<String, Object>> res = jdbcTemplate.queryForList(sql, params);
			if (res != null && !res.isEmpty()) {
				res.forEach(o -> {
					EventContext ctx = new EventContext(topic);
					ctx.setData(paramName, o);
					try {
						eventBus.publish(ctx);
					} catch (Exception e) {
						log.error("发送数据抽取事件失败", e);
					}
				});
			}
		}
		EventContext ctx = new EventContext(topic);
		ctx.setData(Params.PARAM_EXTRACT_DONE, true);
		try {
			eventBus.publish(ctx);
		} catch (Exception e) {
			log.error("发送数据抽取事件失败", e);
		}
	}
}
